/**
	httpproxy.cpp
	~~~~~~~~~~~~~~~~~~~~~~~~~


	Copyright (c) 2007-2008 <ptthippo@gmail.com>

	Permission is hereby granted, free of charge, to any person obtaining a copy
	of this software and associated documentation files (the "Software"), to deal
	in the Software without restriction, including without limitation the rights
	to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
	copies of the Software, and to permit persons to whom the Software is
	furnished to do so, subject to the following conditions:

	The above copyright notice and this permission notice shall be included in
	all copies or substantial portions of the Software.

	THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
	IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
	FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
	AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
	LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
	OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
	THE SOFTWARE.
*/
#include "../common/config.h"
#include <vector>
#include <boost/assign/list_of.hpp>
#include <boost/algorithm/string.hpp>
#include <boost/bind.hpp>
#include <boost/format.hpp>
#include <boost/foreach.hpp>
#include <boost/lexical_cast.hpp>
#include <boost/tokenizer.hpp>
#include <hippolib/util/base64.hpp>
#include <hippolib/net/nettool.hpp>
#include <hippolib/net/httpheader.hpp>
#include "httpproxy.h"
#include "httpsession.h"

using namespace std;
using namespace log4cxx;
namespace ba = boost::algorithm;
using boost::asio::ip::tcp;
typedef boost::tokenizer<boost::char_separator<char> > Tok;

const string HTTPProxy::PACKAGE_NAME = "XRelayer";
const string HTTPProxy::PACKAGE_VERSION = "0.1";
const string HTTPProxy::DEFAULT_SSL_RESPONSE =
		"HTTP/1.0 200 Connection established\r\nProxy-agent: XRelayer/0.1\r\n\r\n";
const string HTTPProxy::DEFAULT_UNAUTHORIZED_RESPONSE =
		"HTTP/1.0 407 Unauthorized\r\nServer: XRelayer\r\nProxy-Authenticate: Basic realm=\"XRelayer Authorization\"\r\nCache-control: no-cache\r\n\r\n";
const boost::regex HTTPProxy::URL_REGEX =
		boost::regex("([a-zA-Z]+\\://)?([\\.\\-0-9a-zA-Z]+)(\\:[0-9]+)?(.*)");

HTTPProxy HTTPProxy::m_instance;

HTTPProxy::HTTPProxy() :
	m_acceptor(m_ioservice),
	m_proxy_check_thread(NULL),
	m_logger(Logger::getLogger("httpproxy")),
	m_access_logger(Logger::getLogger("access"))
{
}

HTTPProxy::~HTTPProxy()
{
}

HTTPProxy& HTTPProxy::getInstance()
{
	return m_instance;
}

void HTTPProxy::Initialize()
{
	vector<string> db_filenames = m_config.getUserDBFilename();
	BOOST_FOREACH(string fn, db_filenames)
	{
		if(m_usermgr.addUserDBFile(fn) == false)
		{
			LOG4CXX_ERROR(m_logger, string("cannot add user db ") + fn);
		}
	}

	vector<HTTPUser> users = m_config.getUsers();
	BOOST_FOREACH(HTTPUser u, users)
	{
		m_usermgr.addUser(u);
	}
}

void HTTPProxy::StartProxy()
{
	tcp::endpoint endpoint(boost::asio::ip::tcp::v4(), m_config.getListeningPort());

	m_acceptor.open(endpoint.protocol());
	m_acceptor.set_option(boost::asio::ip::tcp::acceptor::reuse_address(true));
	m_acceptor.bind(endpoint);
	m_acceptor.listen();
	
	boost::shared_ptr<tcp::socket> clientsock(new tcp::socket(m_ioservice));
	m_acceptor.async_accept(*clientsock,
							boost::bind(&HTTPProxy::handle_accept,
									this,
									clientsock,
									boost::asio::placeholders::error
									)
							);
	this->start();
	m_proxy_check_thread = new boost::thread(boost::bind(&HTTPProxy::proxy_check_thread, this));

	LOG4CXX_INFO(m_logger, "xrelayer started");
}

void HTTPProxy::StopProxy()
{
	boost::system::error_code ec;
	
	m_acceptor.cancel(ec);
	if(ec)
	{
		LOG4CXX_WARN(m_logger, "socket cancel error: " + ec.message());
	}
	m_acceptor.close(ec);
	if(ec)
	{
		LOG4CXX_WARN(m_logger, "socket close error: " + ec.message());
	}
	
	// this will cause asio service threads to exit
	m_ioservice.stop();

	// this will cause proxy_check_thread() to exit
	stop();
}

void HTTPProxy::run()
{
	vector<boost::shared_ptr<boost::thread> > threads;
	size_t thread_num = m_config.getThreads();
	
	for(size_t i = 0; i < thread_num; i++)
	{
		boost::shared_ptr<boost::thread> thread(new boost::thread(
					boost::bind(&boost::asio::io_service::run, &m_ioservice)));
		threads.push_back(thread);
	}

	for(size_t i = 0; i < threads.size(); i++)
		threads[i]->join();
	
	// Wait 2 seconds for other threads to terminate
	hippolib::thread::sleep(2000);

	delete m_proxy_check_thread;
	m_proxy_check_thread = NULL;

	LOG4CXX_INFO(m_logger, "xrelayer terminated");
	// after exiting this function, calling isAlive() will return false
}

void HTTPProxy::proxy_check_thread()
{
	vector<UpstreamProxy> proxies = m_config.getUpstreamProxies();
	unsigned int sleep_seconds = 601;

	LOG4CXX_DEBUG(m_logger, boost::format("Proxy checking thread starts - %d proxies") %
											proxies.size());

	while(IsStopping() == false)
	{
		if(sleep_seconds > 600)
		{
			sleep_seconds = 0;
			BOOST_FOREACH(UpstreamProxy proxy, proxies)
			{
				string name = proxy.getName();
				string host = proxy.getHost();
				unsigned short port = proxy.getPort();

				ProxyStatus status;
				status.setName(name);

				boost::asio::ip::address_v4 ip;
				try
				{
					ip = boost::asio::ip::address_v4::from_string(host);
					status.setResolvedIp(ip);
				}
				catch(...)
				{
					// treat host as a domain name
					try
					{
						ip = hippolib::nettool::Resolve(host);
						status.setResolvedIp(ip);
					}
					catch(...)
					{
						// cannot resolve the domain name
						LOG4CXX_WARN(m_logger, boost::format("Upstream proxy %s %s:%d cannot be resolved") %
															name % host % port);
						status.setAlive(false);
						m_proxystatus[name] = status;
						continue;
					}
				}

				boost::system::error_code err;
				tcp::endpoint proxy_endpoint(ip, port);
				tcp::socket proxy_sock(m_ioservice);
				proxy_sock.connect(proxy_endpoint, err);

				
				if(!err)
					status.setAlive(true);
				else
				{
					LOG4CXX_WARN(m_logger, boost::format("Upstream proxy %s %s:%d is down") %
														name % host % port);
					status.setAlive(false);
				}
				
				m_proxystatus[name] = status;
			}
		}
		
		hippolib::thread::sleep(1000);
		++sleep_seconds;
	}
}

vector<string> HTTPProxy::SKIP_HEADERS = boost::assign::list_of("connection")("proxy-connection")("keep-alive");
	
void
HTTPProxy::process_client_headers(
		map<string, hippolib::httpheader, hippolib::HeaderKeyComparer> & headers,
		int & retcode)
{
	map<string, hippolib::httpheader, hippolib::HeaderKeyComparer>::iterator iter;
	bool authenticated = false;


	// Do authentication if needed
	iter = headers.find("proxy-authorization");
	if(iter != headers.end())
	{
		if(m_config.isAuthEnabled())
		{
			string value = iter->second.getValue();

			// only support basic authentication
			if(value.size() < 9 || ba::starts_with(value, "Basic ") == false)
			{
				retcode = AUTH_NOTSUPPORT;
				return;
			}

			string encoded_str = value.substr(6);
			string decoded_str = hippolib::base64::Decode(encoded_str);
			size_t colon_pos = decoded_str.find_first_of(':');
			if(colon_pos == string::npos || colon_pos == decoded_str.size()-1)
			{
				retcode = AUTH_NOTSUPPORT;
				return;
			}

			string username = decoded_str.substr(0, colon_pos);
			string password = decoded_str.substr(colon_pos+1);
			if(m_usermgr.authUser(username, password) == false)
			{
				retcode = AUTH_FAIL;
				return;
			}

			// username and password are matched!!
			authenticated = true;
		}

		// Don't send this header to the remote web server
		headers.erase(iter);
	}

	// remove the skipped headers
	BOOST_FOREACH(string h, SKIP_HEADERS)
	{
		if((iter = headers.find(h)) != headers.end())
			headers.erase(iter);
	}

	if(m_config.isAuthEnabled() && !authenticated)
	{
		retcode = NEED_AUTH;
		return;
	}
	
	retcode = CORRECT;
}

int HTTPProxy::search_acl(
			const boost::asio::ip::address_v4 & srcip,
			const string & host,
			const boost::asio::ip::address_v4 & hostip)
{
	vector<AccessRule> rules = m_config.getAccessRules();
	BOOST_FOREACH(AccessRule rule, rules)
	{
		boost::shared_ptr<hippolib::host> src = rule.getSource();
		boost::shared_ptr<hippolib::host> dst = rule.getDestination();

		if(src && src->contains(srcip) == false)
			continue;

		if(dst && dst->contains(host) == false && dst->contains(hostip) == false)
			continue;

		return rule.getAction();
	}

	return AccessRule::NOTFOUND;
}

UpstreamProxy
HTTPProxy::search_upstream(
			const boost::asio::ip::address_v4 & srcip,
			const string & host,
			const boost::asio::ip::address_v4 & hostip)
{
	vector<UpstreamRule> rules = m_config.getUpstreamRules();

	BOOST_FOREACH(UpstreamRule rule, rules)
	{
		boost::shared_ptr<hippolib::host> src = rule.getSource();
		boost::shared_ptr<hippolib::host> dst = rule.getDestination();

		if(src && src->contains(srcip) == false)
			continue;

		if(dst && dst->contains(host) == false && dst->contains(hostip) == false)
			continue;

		// search the proxy among the pre-defined ones
		string proxyname = rule.getProxyName();
		vector<UpstreamProxy> proxies = m_config.getUpstreamProxies();
		BOOST_FOREACH(UpstreamProxy proxy, proxies)
		{
			if(proxyname == proxy.getName())
				return proxy;
		}
	}

	return UpstreamProxy();
}

void
HTTPProxy::handle_accept(
			boost::shared_ptr<tcp::socket> clientsock,
			const boost::system::error_code & error)
{
	boost::shared_ptr<tcp::socket> newclientsock(new tcp::socket(m_ioservice));
	m_acceptor.async_accept(*newclientsock, boost::bind(&HTTPProxy::handle_accept, this,
												newclientsock, boost::asio::placeholders::error));

	if(error)
	{
		LOG4CXX_WARN(m_logger, string("accept client error: ") + error.message());
		return;
	}

	boost::asio::streambuf request_streambuf(m_config.getMaxHeaderBufferSize());
	string line, tmp;
	boost::system::error_code err;

	tcp::endpoint client_endpoint = clientsock->remote_endpoint(err);
	if(err)
	{
		LOG4CXX_WARN(m_logger, "cannot get the endpoint of client: " + err.message());
		return;
	}

	LOG4CXX_DEBUG(m_logger, boost::format("%s:%d is connected") %
							client_endpoint.address().to_string() % client_endpoint.port());

	// read the request headers from the client
    	boost::asio::read_until(*clientsock, request_streambuf, boost::regex("\r\n\r\n"), err);
	if(err)
	{
		LOG4CXX_WARN(m_logger, boost::format("read headers error from %s:%d: %s") %
								client_endpoint.address().to_string() % client_endpoint.port() %
								err.message());
		return;
	}
	istream stream(&request_streambuf);


	
	string http_method, url, http_version, protocol, host, path;
	boost::asio::ip::address_v4 hostip;
	unsigned short port;
	map<string, hippolib::httpheader, hippolib::HeaderKeyComparer> headers;

	// parse the first request line
	getline(stream, line);
	boost::char_separator<char> sep(" \r", "");
	Tok tokens(line, sep);
	Tok::iterator tok_iter = tokens.begin();

	// parse the http method - GET, POST, etc
	if(tok_iter == tokens.end())
	{
		LOG4CXX_WARN(m_logger, "Bad HTTP request, method is missing");
		return;
	}
	else
	{
		http_method = *tok_iter;
		++tok_iter;
	}

	// parse the path
	if(tok_iter == tokens.end())
	{
		LOG4CXX_WARN(m_logger, "Bad HTTP request, path is missing");
		return;
	}
	else
	{
		url = *tok_iter;
		++tok_iter;
	}

	//parse the http version - HTTP/1.0 or HTTP/1.1
	if(tok_iter == tokens.end())
	{
		LOG4CXX_WARN(m_logger, "Bad HTTP request, version is missing");
		return;
	}
	else
	{
		http_version = *tok_iter;
		++tok_iter;
	}

	// parse the request headers
	while(getline(stream, line) && line != "\r")
	{
		hippolib::httpheader h;

		try
		{
			h = hippolib::httpheader::parseString(line);
		}
		catch(...)
		{
			LOG4CXX_WARN(m_logger, "cannot parse header-line " + line);
			return;
		}
		
		headers.insert(make_pair(h.getName(), h));
	}

	// parse the host and port from the path string
	port = 0;
	boost::cmatch what;
	if(boost::regex_match(url.c_str(), what, URL_REGEX))
	{
		protocol.assign(what[1].first, what[1].second);
		host.assign(what[2].first, what[2].second);
		tmp.assign(what[3].first, what[3].second);
		path.assign(what[4].first, what[4].second);

		if(tmp != "")
		{
			tmp = tmp.substr(1);
			port = boost::lexical_cast<unsigned short>(tmp);
		}
		
		if(protocol == "")
		{
			if(port == 443)
				protocol = "https";
			else
				protocol = "http";
		}
		else
		{
			protocol = protocol.substr(0, protocol.size()-3);
		}
	}
	else
	{
		LOG4CXX_WARN(m_logger, "parse url " + url + " error");
		return;
	}

	// try to resolve the address of the host
	try
	{
		if(port == 0)
		{
			tcp::endpoint p = hippolib::nettool::ResolveTcpEndpoint(host, protocol);
			hostip = p.address().to_v4();
			port = p.port();
		}
		else
			hostip = hippolib::nettool::Resolve(host);

		LOG4CXX_DEBUG(m_logger, host + " is resolved as " + hostip.to_string() + ":" +
								boost::lexical_cast<string>(port));
	}
	catch(const exception & e)
	{
		LOG4CXX_INFO(m_logger, "cannot resolve host "+ host + ": " + e.what());
		return;
	}

	// match access rules
	if(m_config.isAccessRuleEnabled())
	{
		int ret = search_acl(client_endpoint.address().to_v4(), host, hostip);
		if(ret == AccessRule::NOTFOUND)
			ret = m_config.getAccessRuleDefaultAction();

		if(ret == AccessRule::DENY)
		{
			LOG4CXX_INFO(m_logger, boost::format("client %s is denied to access %s(%s)") %
				client_endpoint.address().to_string() % host % hostip.to_string());
			return;
		}
	}

	// process the headers of client
	int process_retcode;
	process_client_headers(headers, process_retcode);

	// We want to authenticate clients, send the authentication request to the client
	if(process_retcode == NEED_AUTH || process_retcode == AUTH_FAIL || process_retcode == AUTH_NOTSUPPORT)
	{
		if(process_retcode == AUTH_FAIL || process_retcode == AUTH_NOTSUPPORT)
		{
			LOG4CXX_INFO(m_logger, boost::format("authentication for %s is failed, code: %d") %
							client_endpoint.address().to_string() % process_retcode);
		}
		
		boost::asio::write(*clientsock, boost::asio::buffer(DEFAULT_UNAUTHORIZED_RESPONSE),
						boost::asio::transfer_all(), err);
		if(err)
		{
			LOG4CXX_WARN(m_logger, boost::format("send unauthorized response to %s:%d error: %d, %s") %
									client_endpoint.address().to_string() % client_endpoint.port() %
									err.value() % err.message());
		}
		clientsock->shutdown(tcp::socket::shutdown_both, err);
		clientsock->close(err);
		return;
	}

	// try to search upstream proxy
	UpstreamProxy upstream_proxy = search_upstream(client_endpoint.address().to_v4(), host, hostip);
	if(upstream_proxy.getPort() != 0)
	{
		// check the status of the upstream proxy
		PXYSTS_ITER pxy_iter = m_proxystatus.find(upstream_proxy.getName());
		if(pxy_iter != m_proxystatus.end())
		{
			if(pxy_iter->second.isAlive() == true)
			{
				hostip = pxy_iter->second.getResolvedIp();
				port = upstream_proxy.getPort();
			}
			else
				upstream_proxy = UpstreamProxy();
		}
		else
			upstream_proxy = UpstreamProxy();
	}


	// try to connect to the server or the upstream proxy
	tcp::endpoint server_endpoint(hostip, port);
	boost::shared_ptr<tcp::socket> serversock(new tcp::socket(m_ioservice));
	serversock->connect(server_endpoint, err);
	if(err)
	{
		if(upstream_proxy.getPort() == 0)
		{
			LOG4CXX_INFO(m_access_logger, boost::format("%s cannot %s %s(%s):%d/%s, error: %s") %
								client_endpoint.address().to_string() % http_method %
								host % hostip.to_string() % port % url % err.message());
		}
		else
		{
			LOG4CXX_INFO(m_access_logger, boost::format("%s cannot %s %s/%s via proxy %s(%s:%d), error: %s") %
								client_endpoint.address().to_string() % http_method %
								host % url % upstream_proxy.getName() % hostip.to_string() % port % err.message());
		}
		return;
	}
	else
	{
		if(upstream_proxy.getPort() == 0)
		{
			LOG4CXX_INFO(m_access_logger, boost::format("%s %s %s(%s):%d/%s") %
								client_endpoint.address().to_string() % http_method %
								host % hostip.to_string() % port % url);
		}
		else
		{
			LOG4CXX_INFO(m_access_logger, boost::format("%s %s %s/%s via proxy %s(%s:%d)") %
								client_endpoint.address().to_string() % http_method %
								host % url % upstream_proxy.getName() % hostip.to_string() % port);
		}
	}

	HTTPSession *session = new HTTPSession(clientsock, serversock);
	if(upstream_proxy.getPort() == 0)
	{
		session->Initialize(http_method, protocol, host, path, http_version, headers,
						m_config.getHostname(), false, request_streambuf, err);
	}
	else
	{
		session->Initialize(http_method, protocol, host, path, http_version, headers,
						m_config.getHostname(), true, request_streambuf, err);
	}
	
	if(err)
	{
		LOG4CXX_INFO(m_logger, boost::format("initialize from %s:%d to %s:%d error: %s") %
					client_endpoint.address().to_string() % client_endpoint.port() %
					server_endpoint.address().to_string() % server_endpoint.port() % err.message());
		delete session;
		return;
	}

	//session->start();

}

